package org.jetlinks.pro.messaging.rabbitmq.impl;

import org.jetlinks.pro.datasource.CommandHandler;
import org.jetlinks.pro.datasource.DataSource;
import org.jetlinks.pro.messaging.rabbitmq.*;
import org.jetlinks.pro.messaging.rabbitmq.command.CreateConsumer;
import org.jetlinks.pro.messaging.rabbitmq.command.CreateProducer;
import reactor.core.publisher.Mono;

public class CreateProducerHandler implements CommandHandler<CreateProducer, Mono<RabbitMQProducer>> {

    @Override
    public Mono<RabbitMQProducer> handle(DataSource dataSource, CreateProducer command) {
        ReactorRabbitMQProducer consumer = new ReactorRabbitMQProducer(
            command.getProperties() == null
                ? dataSource
                .unwrap(RabbitMQDataSource.class)
                .getConfig()
                : command.getProperties());

        consumer.setBatch(command.isBatch());
        consumer.setMaxBatchSize(command.getBatchSize());
        consumer.setMaxBatchTimeout(command.getBatchTimeout());
        consumer.init();

        return Mono.just(consumer);
    }
}
